package com.gomro.galaxy.foundation.bi.acquisition.job;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.gomro.galaxy.core.batch.ElasticSearchClzItemWriter;
import com.gomro.galaxy.core.commons.elastic.ElasticSearchRest;
import com.gomro.galaxy.foundation.bi.domain.Constants;
import com.gomro.galaxy.foundation.bi.domain.enumerater.AcquireBatchStatus;
import com.gomro.galaxy.foundation.bi.domain.enumerater.AcquirePlan;
import com.gomro.galaxy.foundation.bi.domain.event.BiAcquireBatchEvent;
import com.gomro.galaxy.foundation.bi.domain.model.AcquireGoodsDetail;
import com.gomro.galaxy.foundation.bi.domain.model.AcquireLastBatch;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.builder.JdbcPagingItemReaderBuilder;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.bus.BusProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.jdbc.core.RowMapper;

import javax.sql.DataSource;
import java.util.*;
//商城信息 工作
@Slf4j
@Configuration
public class GoodsDetailJob {

    private static final String BiAcquireLastName = Constants.BiWarehouseIndexPrefix + AcquireLastBatch.TableName;
    private static final String BiWarehouseName = Constants.BiWarehouseIndexPrefix + AcquireGoodsDetail.TableName;
    private static final String JobName = Constants.JobBeanNamePrefix + AcquireGoodsDetail.TableName;
    private static final String StepName = Constants.StepBeanNamePrefix + AcquireGoodsDetail.TableName;
    private static final String ReaderName = Constants.ReaderBeanNamePrefix + AcquireGoodsDetail.TableName;

    RowMapper<AcquireGoodsDetail> rowMapper = (rs, i) -> AcquireGoodsDetail.builder()
            .id(rs.getLong("id"))
            .place(rs.getString("place"))
            .images(rs.getString("images"))
            .description(rs.getString("description"))
            .technical(rs.getString("technical"))
            .packaging(rs.getString("packaging"))
            .memo(rs.getString("memo"))
            .del(rs.getInt("del") == 1)
            .last(rs.getTimestamp("last") == null ? rs.getTimestamp("time").toLocalDateTime() : rs.getTimestamp("last").toLocalDateTime())
            .time(rs.getTimestamp("time").toLocalDateTime())
            .operator(rs.getInt("operator"))
            .modifiedBy(rs.getString("modified_by"))
            .createdBy(rs.getString("created_by"))
            .deliveryPlace(rs.getString("delivery_place"))
            .version(rs.getInt("version"))
            .build();

    private ApplicationEventPublisher applicationEventPublisher;
    private BusProperties busProperties;
    private JobBuilderFactory jobBuilderFactory;
    private StepBuilderFactory stepBuilderFactory;
    private DataSource dataSource;
    private ElasticSearchRest rest;
    private RestHighLevelClient client;
    private ObjectMapper mapper;

    public GoodsDetailJob(ApplicationEventPublisher applicationEventPublisher, BusProperties busProperties, JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory, RestHighLevelClient client,ObjectMapper mapper, @Qualifier("goodsDataSource") DataSource dataSource) {

        this.applicationEventPublisher = applicationEventPublisher;
        this.busProperties = busProperties;
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.dataSource = dataSource;

        this.client = client;
        this.mapper = mapper;
        this.rest = new ElasticSearchRest(client, mapper);
    }

    @Bean(JobName)
    public Job job(@Qualifier(StepName) Step step) {

        return jobBuilderFactory.get(JobName)
                .incrementer(new RunIdIncrementer())
                .flow(step)
                .end()
                .listener(jobExecutionListener)
                .build();
    }

    @Bean(StepName)
    public Step step() {

        return stepBuilderFactory.get(StepName)
                .<AcquireGoodsDetail, AcquireGoodsDetail>chunk(100)
                .reader(reader())
                .processor(processor())
                .writer(writer())
                .build();
    }

    @Bean(ReaderName)
    @StepScope
    public JdbcPagingItemReader<AcquireGoodsDetail> reader() {

        // 获取最后更新时间
        AcquireLastBatch acquireLastBatch = rest.documentGet(BiAcquireLastName, AcquireGoodsDetail.TableName, AcquireLastBatch.class);
        Date last = null;
        if (acquireLastBatch != null && acquireLastBatch.getLast() != null) last = acquireLastBatch.getLast();

        String where = null;
        Map<String, Object> parameterValues = new HashMap<>();
        if (last != null) {
            where = "where (last is not null and last >= :last) or (time >= :time)";
            parameterValues.put("last", last);
            parameterValues.put("time", last);
        }

        return new JdbcPagingItemReaderBuilder<AcquireGoodsDetail>()
                .name(ReaderName)
                .dataSource(dataSource)
                .pageSize(100)
                .selectClause("select *")
                .fromClause("from " + AcquireGoodsDetail.TableName)
                .whereClause(where)
                .parameterValues(parameterValues)
                .sortKeys(new LinkedHashMap<String, Order>() {


                    private static final long serialVersionUID = 2845080114455775121L;

                    {
                        put("id", Order.ASCENDING);
                    }
                })
                .rowMapper(rowMapper)
                .build();
    }

    public ElasticSearchClzItemWriter<AcquireGoodsDetail> writer() {

        return new ElasticSearchClzItemWriter<>(client,mapper, Constants.BiWarehouseIndexPrefix + AcquireGoodsDetail.TableName);
    }

    public ItemProcessor<AcquireGoodsDetail, AcquireGoodsDetail> processor() {

        return item -> item;
    }

    private JobExecutionListener jobExecutionListener = new JobExecutionListener() {

        @Override
        public void beforeJob(@NonNull JobExecution jobExecution) {

            ElasticSearchRest rest = new ElasticSearchRest(client,mapper);

            // 初始化数据索引
            if (!rest.indexExists(BiWarehouseName)) {

                rest.indexCreate(BiWarehouseName);
                rest.indexMappingPut(BiWarehouseName, "acquisition/" + BiWarehouseName + ".json", true);
                rest.indexFlush(BiWarehouseName, null, null);
            }

            // 初始化最后更新时间索引
            if (!rest.indexExists(BiAcquireLastName)) {

                rest.indexCreate(BiAcquireLastName);
                rest.indexMappingPut(BiAcquireLastName, "acquisition/" + BiAcquireLastName + ".json", true);
                rest.indexFlush(BiAcquireLastName, null, null);
            }
        }

        @Override
        public void afterJob(@NonNull JobExecution jobExecution) {

            // 回写最后更新时间
            if (jobExecution.getStatus() == BatchStatus.COMPLETED) {

                rest.documentCreate(Constants.BiWarehouseIndexPrefix + AcquireLastBatch.TableName, AcquireGoodsDetail.TableName, AcquireLastBatch.builder().id(AcquireGoodsDetail.TableName).last(jobExecution.getCreateTime()).build());

                applicationEventPublisher.publishEvent(new BiAcquireBatchEvent(this, busProperties.getId(), "**", AcquirePlan.GoodsDetailPlan, AcquireBatchStatus.JobCompleted));
            }
        }
    };


}
